package com.ruoyi.sjl.service.impl;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import com.google.common.collect.ImmutableMap;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.common.core.web.domain.AjaxResult;
import com.ruoyi.common.redis.service.RedisService;
import com.ruoyi.sjl.handler.RescueWebSocketHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.stereotype.Service;
import com.ruoyi.sjl.mapper.RescueMapper;
import com.ruoyi.sjl.domain.Rescue;
import com.ruoyi.sjl.service.IRescueService;

import javax.annotation.PreDestroy;

/**
 * 救援任务Service业务层处理
 * 
 * @author ruoyi
 * @date 2025-11-13
 */
@Service
@Slf4j
public class RescueServiceImpl implements IRescueService 
{
    @Autowired
    private RescueMapper rescueMapper;
    @Autowired
    private RedisService redisService;
    @Autowired
    private RescueWebSocketHandler rescueWebSocketHandler;
    private static final String RESCUE_STATUS_KEY_PREFIX = "rescue:status:";
    private static final int DELAY_MINUTES = 5;
    private static final String GEO_KEY = "merchant:geo";
    private static final double[] SEARCH_RADII = {1, 3, 10}; // 公里
    // 线程池，每30秒轮询一次商家救援订单状态
    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);

    /**
     * 查询救援任务
     * 
     * @param id 救援任务主键
     * @return 救援任务
     */
    @Override
    public Rescue selectRescueById(Integer id)
    {
        return rescueMapper.selectRescueById(id);
    }

    /**
     * 查询救援任务列表
     * 
     * @param rescue 救援任务
     * @return 救援任务
     */
    @Override
    public List<Rescue> selectRescueList(Rescue rescue)
    {
        return rescueMapper.selectRescueList(rescue);
    }

    /**
     * 新增救援任务
     * 
     * @param rescue 救援任务
     * @return 结果
     */
    @Override
    public int insertRescue(Rescue rescue)
    {
        rescue.setCreateTime(DateUtils.getNowDate());
        return rescueMapper.insertRescue(rescue);
    }

    /**
     * 修改救援任务
     * 
     * @param rescue 救援任务
     * @return 结果
     */
    @Override
    public int updateRescue(Rescue rescue)
    {
        rescue.setUpdateTime(DateUtils.getNowDate());
        return rescueMapper.updateRescue(rescue);
    }

    /**
     * 批量删除救援任务
     * 
     * @param ids 需要删除的救援任务主键
     * @return 结果
     */
    @Override
    public int deleteRescueByIds(Integer[] ids)
    {
        return rescueMapper.deleteRescueByIds(ids);
    }

    /**
     * 删除救援任务信息
     * 
     * @param id 救援任务主键
     * @return 结果
     */
    @Override
    public int deleteRescueById(Integer id)
    {
        return rescueMapper.deleteRescueById(id);
    }
    // 容器销毁时关闭线程池
    @PreDestroy
    public void destroyExecutor() {
        log.info("开始关闭救援轮询线程池");
        if (executor != null && !executor.isShutdown()) {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.warn("线程池关闭超时，强制关闭");
                    executor.shutdownNow();
                }
            } catch (InterruptedException e) {
                log.error("线程池关闭被中断", e);
                executor.shutdownNow();
            }
        }
        log.info("线程池关闭完成");
    }

    /*
    * 监测redis用户救援状态
    * */
    private void startPolling(Integer rescueId,double latitude, double longitude,int preIndex)
    {
        long startTime = System.currentTimeMillis(); // 记录阶段开始时间
        long timeoutMillis = DELAY_MINUTES * 60 * 1000; // 5分钟超时
        Runnable pollingTask = new Runnable() {
            @Override
            public void run() {
                try {
                    Integer cacheObject = redisService.getCacheObject(RESCUE_STATUS_KEY_PREFIX + rescueId.toString());
                    if (cacheObject != null && cacheObject.equals(1)) {
                        log.info("救援订单{}：已接单，终止轮询", rescueId);
                        //已接单
                        return;
                    }
                    // 检查是否超时（5分钟）
                    long currentTime = System.currentTimeMillis();
                    if (currentTime - startTime >= timeoutMillis) {
                        // 超时未接单：推进到下一阶段
                        handleExistMerchant(rescueId, latitude, longitude, preIndex + 1);
                        return; // 终止当前轮询
                    }
                    executor.schedule(this, 30, TimeUnit.SECONDS);
                } catch (Exception e) {
                    log.error("救援订单{}：轮询状态失败", rescueId, e);
                    // 未超时、未接单：继续30秒轮询
                    executor.schedule(this, 30, TimeUnit.SECONDS); // 30秒后继续轮询
                }
            }
        };
        // 启动第一次轮询
        executor.schedule(pollingTask, 0, TimeUnit.SECONDS);
    }

    /*
     * 广播逻辑封装
     * */
    private void executePhase(Integer rescueId,double latitude, double longitude,int preIndex) {
        //1查询该订单状态是否为已接单
        Integer cacheObject = redisService.getCacheObject(RESCUE_STATUS_KEY_PREFIX + rescueId.toString());
        if (cacheObject != null && cacheObject.equals(1)) {
            //已接单
            return;
        }
        if (preIndex < SEARCH_RADII.length) {
            // 获取指定半径内商家
            List<RedisGeoCommands.GeoLocation<String>> merchants = null;
            try {
                merchants = redisService.geoSearch(GEO_KEY, longitude, latitude, SEARCH_RADII[preIndex]);
            } catch (Exception e) {
                log.error("救援订单{}：搜索{}公里内商家失败", rescueId, SEARCH_RADII[preIndex], e);
                // 5秒后重试当前阶段（避免直接失败）
                executor.schedule(() -> executePhase(rescueId, latitude, longitude, preIndex), 5, TimeUnit.SECONDS);
                return;
            }
            // ========== 新增：过滤已通知的商家（核心逻辑） ==========
            String notifiedMerchantsKey = "rescue:notified:" + rescueId; // 记录已通知商家的Redis键
            // 1. 获取该订单已通知过的商家ID集合
            Set<String> notifiedMerchantIds = redisService.getCacheSet(notifiedMerchantsKey);
            if (notifiedMerchantIds == null) {
                notifiedMerchantIds = new HashSet<>();
            }
            //2.2 stream过滤：只保留未通知过的商家
            Set<String> finalNotifiedIds = notifiedMerchantIds; // stream中引用需final
            // 2. 过滤：只保留“未通知过”的商家
            List<RedisGeoCommands.GeoLocation<String>> newMerchants = merchants.stream()
                    .filter(merchant -> !finalNotifiedIds.contains(merchant.getName()))
                    .collect(Collectors.toList());
            // 2.3 提取本次新增的商家ID（用于更新Redis集合）
            Set<String> newMerchantIds = newMerchants.stream()
                    .map(RedisGeoCommands.GeoLocation::getName)
                    .collect(Collectors.toSet());
            if (!newMerchants.isEmpty()) {
                // 去执行存在商家但可能未接单、抢单的情况 通过websocket通知商家
                rescueWebSocketHandler.broadcastMerchantsToNearby(newMerchants,"task_new", ImmutableMap.of(
                        "rescueId", rescueId,
                        "status","待抢单",
                        "message", "有新的订单，请尽快处理"
                ));
                // ========== 新增：记录本次通知的商家ID到Redis ==========
                notifiedMerchantIds.addAll(newMerchantIds); // 旧集合 + 新商家ID
                redisService.setCacheSet(notifiedMerchantsKey, notifiedMerchantIds); // 覆盖存回Redis
                redisService.expire(notifiedMerchantsKey, 24L, TimeUnit.HOURS); // 设24小时过期
                //调用线程监测订单状态是否改变
                startPolling(rescueId, latitude, longitude, preIndex);
            }
            else {
                // 当前半径无商家：直接推进到下一阶段
                handleExistMerchant(rescueId, latitude, longitude, preIndex + 1);
            }
        }
    }
    /*
     * 封装几公里内有商家  1公里5分钟 3公里5分钟 10公里5分钟
     * */
    private void handleExistMerchant(Integer rescueId,double latitude, double longitude,int preIndex) {
        // 防越界：超过10公里则取消订单（preIndex最大为2，preIndex+1=3时触发）
        if (preIndex >= SEARCH_RADII.length) {
            Rescue rescue = rescueMapper.selectRescueById(rescueId);
            if (rescue != null && rescue.getStatus() != 1) {
                rescue.setStatus(5);
                rescue.setUpdateTime(DateUtils.getNowDate());
                rescueMapper.updateRescue(rescue);
                redisService.setCacheObject(RESCUE_STATUS_KEY_PREFIX + rescueId, 5, 24L, TimeUnit.HOURS);
                rescueWebSocketHandler.notifyUser(rescue.getUid().toString(),"rescue_fail", ImmutableMap.of(
                        "rescueId", rescueId,
                        "message", "订单已取消，无商家接单!"
                ));
            }
            return;
        }
        executePhase(rescueId, latitude, longitude, preIndex);
    }
    /*
    * 封装判断几公里内无商家
    * */
        private Integer isNoMerchant(Integer rescueId,double latitude, double longitude) {
        // 按1→3→10公里顺序扫描，找到第一个有商家就停止（不重复扫描）
        for (int preIndex = 0; preIndex < SEARCH_RADII.length; preIndex++) {
            double radius = SEARCH_RADII[preIndex];
            List<RedisGeoCommands.GeoLocation<String>> merchants = redisService.geoSearch(GEO_KEY, longitude, latitude, radius);
            log.info("救援订单{}：{}公里内查询到商家数：{}，商家ID列表：{}",
                    rescueId, radius, merchants.size(),
                    merchants.stream().map(m -> m.getName()).collect(Collectors.toList()));
            if (merchants.size() > 0) {
                //返回第几阶段有商家
                return  preIndex;
            }
        }
        // 所有半径都无商家：取消订单+通知用户
        Rescue rescue = rescueMapper.selectRescueById(rescueId);
        if (rescue != null) {
            // 设置为订单为已取消
            rescue.setStatus(5);
            rescue.setUpdateTime(DateUtils.getNowDate());
            rescueMapper.updateRescue(rescue);
            rescueWebSocketHandler.notifyUser(rescue.getUid().toString(),"rescue_fail", ImmutableMap.of(
                    "rescueId", rescueId,
                    "message", "订单已取消，附近无商家!"
            ));
        }
        return -1;
    }
    /*
    * 用户新增救援
    * */
    @Override
    public AjaxResult addRescue(Rescue rescue){
         if(rescue.getStatus() == null) {
            rescue.setStatus(0);
            rescue.setCreateTime(DateUtils.getNowDate());
            }
         Integer status = rescue.getStatus();

         //保存救援信息
         rescueMapper.insertRescue(rescue);
         Integer rescueId = rescue.getId();
         String statusKey  = RESCUE_STATUS_KEY_PREFIX + rescueId.toString();
         redisService.setCacheObject(statusKey , status, 24L, TimeUnit.HOURS);
         Double latitude = rescue.getLatitude();
         Double longitude = rescue.getLongitude();
         Integer merchantsFound = isNoMerchant(rescueId, latitude, longitude);
         if (merchantsFound>=0) {
             executePhase(rescueId, latitude, longitude, merchantsFound);
             return AjaxResult.success("已发送商家信息");
         }

         return AjaxResult.error("附近无商家");

        }
    }

